package com.navinfo.platform.score.utils;

import com.navinfo.platform.score.pojo.ScoreWeight;
import com.navinfo.platform.score.pojo.ScoreWeightCompiled;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

/**
 *
 * @author CAI
 * @create 2019/03/13
 */

public class BaseInfoBroadcastUtil implements Serializable {
    /**
     * 广播权重+打分公式信息
     *
     * @param jsc JavaSparkContext
     * @param scoreWeightRdd 权重+打分公式信息
     * @return
     */
    public static Broadcast<Map<String, ScoreWeightCompiled>> createScoreWeightBroadcast(JavaSparkContext jsc,
                                                                                                         JavaPairRDD<String, ScoreWeightCompiled> scoreWeightRdd) {
        Map<String, ScoreWeightCompiled> scoreWeight = new HashMap<>(128);
        scoreWeight.putAll(scoreWeightRdd.collectAsMap());
        return jsc.broadcast(scoreWeight);
    }
    /**
     * 广播权重+打分公式信息
     *
     * @param jsc JavaSparkContext
     * @param scoreWeightRdd 权重+打分公式信息
     * @return
     */
    public static Map<String, ScoreWeightCompiled> createScoreWeightMap(JavaSparkContext jsc,
                                                                                         JavaPairRDD<String, ScoreWeightCompiled> scoreWeightRdd) {
        Map<String, ScoreWeightCompiled> scoreWeight = new HashMap<>(128);
        scoreWeight.putAll(scoreWeightRdd.collectAsMap());
        return scoreWeight;
    }
}